#include "MessageConsumer.h"
#include <spdlog/spdlog.h>
#include <chrono>
#include <thread>

using TraderEngine::RedisConnection;
using TraderEngine::RedisPool;

namespace MessageFramework {

MessageConsumer::MessageConsumer(RedisPool* redisPool,
                                 const Config& config,
                                 std::shared_ptr<MessageDispatcher> dispatcher)
    : m_redisPool(redisPool),
      m_config(config),
      m_dispatcher(dispatcher),
      m_running(false) {

    if (!m_redisPool) {
        throw std::invalid_argument("RedisPool cannot be null");
    }

    if (m_config.streamName.empty()) {
        throw std::invalid_argument("streamName cannot be empty");
    }

    if (m_config.consumerGroup.empty()) {
        throw std::invalid_argument("consumerGroup cannot be empty");
    }

    if (m_config.consumerId.empty()) {
        throw std::invalid_argument("consumerId cannot be empty");
    }

    // 创建默认的Dispatcher（如果未提供）
    if (!m_dispatcher) {
        m_dispatcher = std::make_shared<MessageDispatcher>();
    }

    spdlog::info("MessageConsumer created: stream={}, group={}, consumer={}",
                 m_config.streamName, m_config.consumerGroup, m_config.consumerId);
}

MessageConsumer::~MessageConsumer() {
    stop();
}

bool MessageConsumer::createConsumerGroup() {
    auto conn = m_redisPool->getConnection();
    if (!conn) {
        spdlog::error("Failed to get Redis connection");
        return false;
    }

    try {
        // XGROUP CREATE stream group <start_id>
        bool success = conn->xgroupCreate(m_config.streamName,
                                          m_config.consumerGroup,
                                          m_config.groupStartId);

        if (success) {
            spdlog::info("Consumer group '{}' created for stream '{}'",
                        m_config.consumerGroup, m_config.streamName);
        } else {
            spdlog::warn("Consumer group '{}' may already exist",
                        m_config.consumerGroup);
        }

        m_redisPool->returnConnection(conn);
        return true;

    } catch (const std::exception& e) {
        // BUSYGROUP错误表示消费组已存在，这是正常情况
        std::string error = e.what();
        if (error.find("BUSYGROUP") != std::string::npos) {
            spdlog::debug("Consumer group '{}' already exists for stream '{}'",
                         m_config.consumerGroup, m_config.streamName);
            m_redisPool->returnConnection(conn);
            return true;
        }
        spdlog::error("Failed to create consumer group: {}", e.what());
        m_redisPool->returnConnection(conn);
        return false;
    }
}

bool MessageConsumer::shouldProcessMessage(const MessageHeader& header) const {
    // 如果不需要过滤，所有消息都处理
    if (!m_config.filterByConsumerId) {
        return true;
    }

    // 如果consumerId为空，表示广播消息，所有消费者都处理
    if (std::strlen(header.consumerId) == 0) {
        return true;
    }

    // 检查consumerId是否匹配
    return std::strcmp(header.consumerId, m_config.consumerId.c_str()) == 0;
}

std::vector<std::pair<Message, std::string>> MessageConsumer::readMessages(std::shared_ptr<RedisConnection> conn) {
    std::vector<std::pair<Message, std::string>> messages;

    try {
        // XREADGROUP GROUP group consumer BLOCK ms COUNT count STREAMS stream >
        auto streamMessages = conn->xreadgroup(
            m_config.consumerGroup,
            m_config.consumerId,
            m_config.streamName,
            ">",  // 读取未消费的消息
            m_config.batchSize,
            m_config.blockMs
        );

        for (const auto& streamMsg : streamMessages) {
            // 从Redis字段构建MessageHeader
            // 将unordered_map转换为map
            std::map<std::string, std::string> fieldsMap(streamMsg.fields.begin(), streamMsg.fields.end());
            MessageHeader header = MessageHeader::fromRedisFields(fieldsMap);

            // 获取body字段
            auto bodyIt = streamMsg.fields.find("body");
            std::string body = (bodyIt != streamMsg.fields.end()) ? bodyIt->second : "";

            // 构建Message，同时保存Stream ID
            Message msg(header, body);
            messages.push_back(std::make_pair(msg, streamMsg.id));

            m_stats.messagesReceived++;
        }

    } catch (const std::exception& e) {
        spdlog::error("Failed to read messages: {}", e.what());
    }

    return messages;
}

bool MessageConsumer::processMessage(std::shared_ptr<RedisConnection> conn,
                                     const Message& message,
                                     const std::string& streamId) {
    // 1. 检查是否应该处理此消息
    if (!shouldProcessMessage(message.header)) {
        spdlog::debug("Message {} filtered (consumerId mismatch)", message.header.messageId);
        m_stats.messagesFiltered++;
        ackMessage(conn, streamId);  // 仍然需要ACK，避免阻塞其他消息
        return true;
    }

    // 2. 处理消息
    int result = -1;

    try {
        if (m_messageCallback) {
            // 使用回调函数处理
            result = m_messageCallback(message);
        } else if (m_dispatcher) {
            // 使用Dispatcher处理
            Message mutableMsg = message;  // Dispatcher需要可修改的消息（解密）
            auto dispatchResult = m_dispatcher->dispatch(mutableMsg);
            result = dispatchResult.code;

            if (result != 0) {
                spdlog::warn("Message {} processing failed: {}",
                           message.header.messageId, dispatchResult.error);
            }
        }

        if (result == 0) {
            m_stats.messagesProcessed++;
        } else {
            m_stats.messagesFailed++;
        }

    } catch (const std::exception& e) {
        spdlog::error("Exception while processing message {}: {}",
                     message.header.messageId, e.what());
        m_stats.messagesFailed++;
        result = -1;
    }

    // 3. 确认消息（即使处理失败也ACK，避免重复处理）
    // 注意：如果需要重试机制，这里应该根据result决定是否ACK
    return ackMessage(conn, streamId);
}

bool MessageConsumer::ackMessage(std::shared_ptr<RedisConnection> conn, const std::string& streamId) {
    try {
        int acked = conn->xack(m_config.streamName, m_config.consumerGroup, {streamId});

        if (acked > 0) {
            spdlog::debug("Message {} acknowledged", streamId);
            return true;
        } else {
            spdlog::warn("Failed to acknowledge message {}", streamId);
            return false;
        }

    } catch (const std::exception& e) {
        spdlog::error("Exception while acknowledging message {}: {}", streamId, e.what());
        return false;
    }
}

void MessageConsumer::consumerLoop() {
    spdlog::info("Consumer loop started: {}", m_config.consumerId);

    auto conn = m_redisPool->getConnection();
    if (!conn) {
        spdlog::error("Failed to get Redis connection for consumer loop");
        return;
    }

    while (m_running) {
        try {
            // 读取消息
            auto messages = readMessages(conn);

            // 处理每条消息
            for (const auto& msgPair : messages) {
                if (!m_running) break;

                // 使用正确的Stream ID进行ACK
                processMessage(conn, msgPair.first, msgPair.second);
            }

        } catch (const std::exception& e) {
            spdlog::error("Exception in consumer loop: {}", e.what());
            std::this_thread::sleep_for(std::chrono::seconds(1));
        }
    }

    m_redisPool->returnConnection(conn);
    spdlog::info("Consumer loop stopped: {}", m_config.consumerId);
}

void MessageConsumer::checkPendingMessages() {
    auto conn = m_redisPool->getConnection();
    if (!conn) {
        spdlog::error("Failed to get Redis connection for pending check");
        return;
    }

    try {
        // XPENDING stream group - + count
        auto pendingMessages = conn->xpending(
            m_config.streamName,
            m_config.consumerGroup,
            "-", "+",
            100  // 最多检查100条
        );

        for (const auto& pending : pendingMessages) {
            // 检查是否超时
            if (pending.elapsedMs > m_config.pendingTimeout * 1000) {
                spdlog::warn("Reclaiming pending message {} from consumer {}",
                           pending.id, pending.consumer);

                // XCLAIM stream group consumer min-idle-time id
                try {
                    conn->xclaim(
                        m_config.streamName,
                        m_config.consumerGroup,
                        m_config.consumerId,
                        m_config.pendingTimeout * 1000,
                        {pending.id}
                    );

                    m_stats.pendingReclaimed++;

                } catch (const std::exception& e) {
                    spdlog::error("Failed to claim message {}: {}", pending.id, e.what());
                }
            }
        }

    } catch (const std::exception& e) {
        spdlog::error("Exception while checking pending messages: {}", e.what());
    }

    m_redisPool->returnConnection(conn);
}

void MessageConsumer::pendingLoop() {
    spdlog::info("Pending check loop started");

    while (m_running) {
        checkPendingMessages();

        // 等待下一次检查
        for (int i = 0; i < m_config.pendingCheckInterval && m_running; ++i) {
            std::this_thread::sleep_for(std::chrono::seconds(1));
        }
    }

    spdlog::info("Pending check loop stopped");
}

void MessageConsumer::start() {
    if (m_running) {
        spdlog::warn("MessageConsumer already running");
        return;
    }

    // 创建Consumer Group（如果需要）
    if (m_config.autoCreateGroup) {
        createConsumerGroup();
    }

    m_running = true;

    // 启动消费者线程
    m_consumerThread = std::thread(&MessageConsumer::consumerLoop, this);

    // 启动Pending检查线程
    m_pendingThread = std::thread(&MessageConsumer::pendingLoop, this);

    spdlog::info("MessageConsumer started: {}", m_config.consumerId);
}

void MessageConsumer::stop() {
    if (!m_running) {
        return;
    }

    spdlog::info("Stopping MessageConsumer: {}", m_config.consumerId);

    m_running = false;

    // 等待线程结束
    if (m_consumerThread.joinable()) {
        m_consumerThread.join();
    }

    if (m_pendingThread.joinable()) {
        m_pendingThread.join();
    }

    spdlog::info("MessageConsumer stopped: {}", m_config.consumerId);
}

void MessageConsumer::setMessageCallback(MessageCallback callback) {
    m_messageCallback = callback;
}

void MessageConsumer::registerHandler(MessageHandlerPtr handler) {
    if (m_dispatcher && handler) {
        m_dispatcher->registerHandler(handler->getMessageType(), handler);
    }
}

void MessageConsumer::updateConfig(const Config& config) {
    m_config = config;
    spdlog::info("MessageConsumer config updated");
}

MessageConsumer::Stats MessageConsumer::getStats() const {
    return m_stats;
}

} // namespace MessageFramework
